package org.zoomdev.zoom;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.sql.*;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import javax.sql.DataSource;

public class OnlineDataMigrationService {

    // 迁移状态监听器接口
    public interface MigrationListener {
        void onStateChanged(MigrationState newState);
        void onProgressUpdate(int progress);
        void onError(Throwable exception);
    }

    // 迁移配置类
    public static class MigrationConfig {
        private int batchSize = 500;
        private int maxRetries = 3;
        private Duration maxPauseDuration = Duration.ofMinutes(10);
        private Duration dualWritePeriod = Duration.ofHours(1);
        private Duration lockDuration = Duration.ofMillis(500);
        private double maxDataDiscrepancy = 0.01; // 1%误差
        private String tmpDir;

        public int getBatchSize() {
            return batchSize;
        }

        public void setBatchSize(int batchSize) {
            this.batchSize = batchSize;
        }

        public int getMaxRetries() {
            return maxRetries;
        }

        public void setMaxRetries(int maxRetries) {
            this.maxRetries = maxRetries;
        }

        public Duration getMaxPauseDuration() {
            return maxPauseDuration;
        }

        public void setMaxPauseDuration(Duration maxPauseDuration) {
            this.maxPauseDuration = maxPauseDuration;
        }

        public Duration getDualWritePeriod() {
            return dualWritePeriod;
        }

        public void setDualWritePeriod(Duration dualWritePeriod) {
            this.dualWritePeriod = dualWritePeriod;
        }

        public Duration getLockDuration() {
            return lockDuration;
        }

        public void setLockDuration(Duration lockDuration) {
            this.lockDuration = lockDuration;
        }

        public double getMaxDataDiscrepancy() {
            return maxDataDiscrepancy;
        }

        public void setMaxDataDiscrepancy(double maxDataDiscrepancy) {
            this.maxDataDiscrepancy = maxDataDiscrepancy;
        }

        public String getTmpDir() {
            return tmpDir;
        }

        public void setTmpDir(String tmpDir) {
            this.tmpDir = tmpDir;
        }

        // Getters and setters with builder pattern
    }

    // 核心迁移器
    public static class DataMigrator implements Runnable {
        private final String taskId;
        private final Set<String> merchantIds;
        private final String sourceTable;
        private final String targetTable;
        private final MigrationConfig config;
        private volatile MigrationState state = MigrationState.PREPARING;
        private final List<MigrationListener> listeners = new CopyOnWriteArrayList<>();

        private final AtomicInteger progress = new AtomicInteger(0);
        private volatile boolean paused = false;
        private volatile boolean cancelled = false;
        private DataSource dataSource;
        private LightLockManager lockManager = new LightLockManager();

        public DataMigrator(String taskId, Set<String> merchantIds, String sourceTable,
                            String targetTable, MigrationConfig config) {
            this.taskId = taskId;
            this.merchantIds = Collections.unmodifiableSet(new HashSet<>(merchantIds));
            this.sourceTable = sourceTable;
            this.targetTable = targetTable;
            this.config = config;
        }

        public void addListener(MigrationListener listener) {
            listeners.add(listener);
        }

        public void pause() {
            this.paused = true;
        }

        public void resume() {
            this.paused = false;
            synchronized (this) {
                this.notifyAll();
            }
        }

        public void cancel() {
            this.cancelled = true;
        }

        public MigrationState getState() {
            return state;
        }

        public int getProgress() {
            return progress.get();
        }

        private void setState(MigrationState newState) {
            this.state = newState;
            listeners.forEach(listener -> listener.onStateChanged(newState));
        }

        private void updateProgress(int newProgress) {
            this.progress.set(newProgress);
            listeners.forEach(listener -> listener.onProgressUpdate(newProgress));
        }

        @Override
        public void run() {
            try {
                executeMigration();
            } catch (MigrationException e) {
                rollbackMigration();
                listeners.forEach(listener -> listener.onError(e));
            } catch (Exception e) {
                setState(MigrationState.ROLLBACK);
                listeners.forEach(listener -> listener.onError(e));
            }
        }

        private void executeMigration() {
            // 1. 准备阶段
            prepareMigration();
            setState(MigrationState.PRE_MIGRATION);

            // 2. 预迁移历史数据
            migrateHistoricalData();

            // 3. 开启双写
            setState(MigrationState.DUAL_WRITING);
            enableDualWriting();

            // 4. 等待双写周期
            waitForDualWritePeriod();

            // 5. 数据验证
            setState(MigrationState.VERIFYING);
            verifyDataConsistency();

            // 6. 切换路由
            setState(MigrationState.SWITCHING);
            switchRouting();

            // 7. 清理旧数据
            setState(MigrationState.CLEANING);
            cleanupData();

            // 8. 完成
            setState(MigrationState.COMPLETED);
        }

        private void prepareMigration() {
            try (Connection conn = dataSource.getConnection()) {
                // 创建新表
                createNewTable(conn);

                // 在元数据表中创建迁移记录
                createMigrationRecord(conn);
            } catch (SQLException e) {
                throw new MigrationException("Preparation failed", e);
            }
        }

        private void createNewTable(Connection conn) throws SQLException {
            // 创建新表（与源表结构相同）
            String createTableSQL = "CREATE TABLE " + targetTable + " LIKE " + sourceTable;

            // 添加迁移版本列（可选）
            if (!tableColumnExists(conn, targetTable, "migration_version")) {
                createTableSQL += ", ADD COLUMN migration_version BIGINT DEFAULT 0";
            }

            try (PreparedStatement stmt = conn.prepareStatement(createTableSQL)) {
                stmt.executeUpdate();
            }

            // 为迁移优化索引（动态创建索引）
            createMigrationIndexes(conn, targetTable);
        }

        // 为迁移创建优化的多列索引
        private void createMigrationIndexes(Connection conn, String tableName) throws SQLException {
            String indexSQL = String.format(
                    "CREATE INDEX idx_%s_migration ON %s (merchant_id, id)",
                    UUID.randomUUID().toString().substring(0, 8),
                    tableName
            );

            try (PreparedStatement stmt = conn.prepareStatement(indexSQL)) {
                stmt.execute();
            }
        }

        // 检查列是否存在
        private boolean tableColumnExists(Connection conn, String table, String column) {
            final String query = "SELECT 1 FROM information_schema.COLUMNS " +
                    "WHERE TABLE_NAME = ? AND COLUMN_NAME = ?";
            try (PreparedStatement stmt = conn.prepareStatement(query)) {
                stmt.setString(1, table);
                stmt.setString(2, column);
                try (ResultSet rs = stmt.executeQuery()) {
                    return rs.next();
                }
            } catch (SQLException e) {
                return false;
            }
        }
        private void createMigrationRecord(Connection conn) throws SQLException {
            // 准备迁移元数据
            String migrateId = UUID.randomUUID().toString();
            LocalDateTime startTime = LocalDateTime.now();

            // 创建迁移记录
            String insertSQL = "INSERT INTO migration_metadata " +
                    "(task_id, source_table, target_table, merchant_ids, " +
                    "start_time, status, progress) " +
                    "VALUES (?, ?, ?, ?, ?, ?, ?)";

            try (PreparedStatement stmt = conn.prepareStatement(insertSQL)) {
                stmt.setString(1, migrateId);
                stmt.setString(2, sourceTable);
                stmt.setString(3, targetTable);
                stmt.setString(4, String.join(",", merchantIds));  // "MER001,MER002,..."
                stmt.setTimestamp(5, Timestamp.valueOf(startTime));
                stmt.setString(6, MigrationState.PREPARING.name());
                stmt.setInt(7, 0);  // 初始进度

                stmt.executeUpdate();
            }

            // 添加分区锁记录（防止重复迁移）
            createPartitionLock(conn, migrateId);
        }

        // 创建分区锁防止并发冲突
        private void createPartitionLock(Connection conn, String migrateId) throws SQLException {
            String lockSQL = "INSERT INTO partition_locks " +
                    "(resource_type, resource_id, lock_owner, expiration) " +
                    "VALUES (?, ?, ?, ?)";

            LocalDateTime expiration = LocalDateTime.now().plusHours(24);

            try (PreparedStatement stmt = conn.prepareStatement(lockSQL)) {
                stmt.setString(1, "MERCHANT_MIGRATION");
                stmt.setString(2, sourceTable + "_" + targetTable);
                stmt.setString(3, migrateId);
                stmt.setTimestamp(4, Timestamp.valueOf(expiration));
                stmt.executeUpdate();
            }
        }

        private void migrateHistoricalData() {
            int totalBatches = calculateTotalBatches();
            AtomicInteger processedBatches = new AtomicInteger(0);

            // 并行迁移不同商户（控制并发度）
            ExecutorService executor = Executors.newFixedThreadPool(
                    Math.min(merchantIds.size(), Runtime.getRuntime().availableProcessors())
            );

            List<Future<?>> futures = merchantIds.stream().map(merchantId ->
                    executor.submit(() -> migrateMerchantData(merchantId, processedBatches, totalBatches))
            ).collect(Collectors.toList());

            // 等待所有迁移任务完成
            for (Future<?> future : futures) {
                try {
                    future.get(); // 等待单个任务完成
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new MigrationException("Migration interrupted", e);
                } catch (ExecutionException e) {
                    throw new MigrationException("Merchant migration failed", e.getCause());
                }
            }

            executor.shutdown();
        }


        private List<Map<String, Object>> fetchBatchWithCursor(
                Connection conn, String merchantId, long lastId, int batchSize) throws SQLException
        {
            String cursorSQL = String.format(
                    "SELECT * FROM %s WHERE merchant_id = ? AND id > ? ORDER BY id LIMIT ?",
                    sourceTable
            );

            List<Map<String, Object>> batch = new ArrayList<>();

            try (PreparedStatement stmt = conn.prepareStatement(cursorSQL)) {
                stmt.setString(1, merchantId);
                stmt.setLong(2, lastId);
                stmt.setInt(3, batchSize);

                try (ResultSet rs = stmt.executeQuery()) {
                    ResultSetMetaData metaData = rs.getMetaData();
                    int columnCount = metaData.getColumnCount();

                    while (rs.next()) {
                        Map<String, Object> row = new LinkedHashMap<>();
                        for (int i = 1; i <= columnCount; i++) {
                            String colName = metaData.getColumnName(i);
                            row.put(colName, rs.getObject(i));
                        }
                        batch.add(row);
                    }
                }
            }
            return batch;
        }

        private void insertBatch(Connection conn, List<Map<String, Object>> batch, String merchantId)
                throws SQLException
        {
            if (batch.isEmpty()) return;

            // 动态生成插入SQL（仅需要插入的列）
            String insertSQL = generateInsertSQL(conn, targetTable, batch.get(0).keySet());

            try (PreparedStatement stmt = conn.prepareStatement(insertSQL)) {
                for (Map<String, Object> row : batch) {
                    int paramIndex = 1;
                    for (String column : row.keySet()) {
                        setParameter(stmt, paramIndex++, row.get(column));
                    }
                    stmt.addBatch();
                }

                // 批量执行
                int[] results = stmt.executeBatch();

                // 验证插入数量
                int insertedCount = (int) Arrays.stream(results).filter(r -> r == 1).count();
                if (insertedCount != batch.size()) {
                    conn.rollback();
                    throw new MigrationException(
                            "Insert mismatch: expected " + batch.size() + ", inserted " + insertedCount
                    );
                }
            }
        }

        private void migrateMerchantData(String merchantId, AtomicInteger processedBatches, int totalBatches) {
            long maxId = getInitialMaxId(merchantId);

            while (true) {
                checkPausedOrCancelled(); // 检查是否暂停或取消

                try (Connection srcConn = dataSource.getConnection();
                     Connection targetConn = dataSource.getConnection()) {

                    // 开启事务保证批量操作的原子性
                    srcConn.setAutoCommit(false);
                    targetConn.setAutoCommit(false);

                    // 获取一批数据（使用游标提高性能）
                    List<Map<String, Object>> batch = fetchBatchWithCursor(
                            srcConn, merchantId, maxId, config.getBatchSize()
                    );

                    if (batch.isEmpty()) break;

                    // 插入到目标表
                    insertBatch(targetConn, batch, merchantId);

                    // 更新最大ID
                    maxId = getMaxId(batch);

                    // 更新进度
                    int currentProgress = processedBatches.incrementAndGet();
                    updateProgress(calculateProgress(currentProgress, totalBatches));

                    // 提交事务
                    srcConn.commit();
                    targetConn.commit();

                } catch (SQLException e) {
                    throw new MigrationException("Batch migration failed for merchant: " + merchantId, e);
                }
            }
        }

        private void enableDualWriting() {
            // 更新所有相关商户的状态
            updateMerchantStatus(merchantIds, "DUAL_WRITE");
        }

        private void waitForDualWritePeriod() {
            LocalDateTime endTime = LocalDateTime.now().plus(config.getDualWritePeriod());
            while (LocalDateTime.now().isBefore(endTime)) {
                // 暂停检查
                checkPausedOrCancelled();

                // 每10秒检查一次
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }

        private void verifyDataConsistency() {
            for (String merchantId : merchantIds) {
                double discrepancy = calculateDataDiscrepancy(merchantId);
                if (discrepancy > config.maxDataDiscrepancy) {
                    throw new MigrationException("Data discrepancy too high: " + discrepancy);
                }
            }
        }


        private boolean verifyFinalConsistency(String merchantId) {
            // 在双写期间已经校验过，这里只需要比较最后一次双写至今的增量数据
            Timestamp switchTime = new Timestamp(System.currentTimeMillis());

            try (Connection conn = dataSource.getConnection()) {
                // 获取源表在双写开启后的最后记录ID
                Long maxSourceId = getMaxIdInTableAfterDualWrite(conn, sourceTable, merchantId);
                Long maxTargetId = getMaxIdInTableAfterDualWrite(conn, targetTable, merchantId);

                // 两个表的最大ID应该相同
                return maxSourceId.equals(maxTargetId);
            } catch (SQLException e) {
                throw new MigrationException("Final consistency check failed", e);
            }
        }

        private Long getMaxIdInTableAfterDualWrite(Connection conn, String table, String merchantId) throws SQLException {
            String sql = "SELECT MAX(id) FROM " + table +
                    " WHERE merchant_id = ? AND created_time >= ?";

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);
                stmt.setTimestamp(2, Timestamp.valueOf(dualWriteStartTime));

                try (ResultSet rs = stmt.executeQuery()) {
                    return rs.next() ? rs.getLong(1) : 0L;
                }
            }
        }
        private void switchRouting() {
            // 对每个商户进行短暂锁定
            for (String merchantId : merchantIds) {
                try {
                    lockManager.acquireLightLock(merchantId, config.lockDuration);

                    // 最后检查数据一致性
                    if (!verifyFinalConsistency(merchantId)) {
                        throw new MigrationException("Final consistency check failed");
                    }

                    // 更新路由表
                    updateRoutingTable(merchantId, targetTable);

                } finally {
                    lockManager.releaseLightLock(merchantId);
                }
            }
        }

        private double calculateDataDiscrepancy(String merchantId) {
            // 比较记录总数
            long sourceCount = countRecords(sourceTable, merchantId);
            long targetCount = countRecords(targetTable, merchantId);
            double discrepancy = Math.abs(sourceCount - targetCount) / (double) sourceCount;

            // 差异在容忍范围内，可以接受
            if (discrepancy <= config.maxDataDiscrepancy) {
                return discrepancy;
            }

            // 详细检查数据内容
            double contentDiscrepancy = compareDataContent(merchantId);
            return Math.max(discrepancy, contentDiscrepancy);
        }

        private long countRecords(String table, String merchantId) {
            String sql = "SELECT COUNT(*) FROM " + table + " WHERE merchant_id = ?";

            try (Connection conn = dataSource.getConnection();
                 PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);
                try (ResultSet rs = stmt.executeQuery()) {
                    return rs.next() ? rs.getLong(1) : 0;
                }
            } catch (SQLException e) {
                throw new MigrationException("Count failed for table: " + table, e);
            }
        }

        private double compareDataContent(String merchantId) {
            // 对关键字段进行校验和比较
            String sourceSql = "SELECT COUNT(*), SUM(CRC32(amount)), SUM(CRC32(order_time)) " +
                    "FROM " + sourceTable + " WHERE merchant_id = ?";
            String targetSql = "SELECT COUNT(*), SUM(CRC32(amount)), SUM(CRC32(order_time)) " +
                    "FROM " + targetTable + " WHERE merchant_id = ?";

            try (Connection conn = dataSource.getConnection()) {
                // 获取源表校验和
                Object[] sourceSums = executeChecksumQuery(conn, sourceSql, merchantId);

                // 获取目标表校验和
                Object[] targetSums = executeChecksumQuery(conn, targetSql, merchantId);

                // 计算差异率
                if (sourceSums[0].equals(targetSums[0]) &&
                        sourceSums[1].equals(targetSums[1]) &&
                        sourceSums[2].equals(targetSums[2])) {
                    return 0.0; // 完全一致
                }

                long sourceTotalChecksum = (long) sourceSums[1] + (long) sourceSums[2];
                long targetTotalChecksum = (long) targetSums[1] + (long) targetSums[2];
                return Math.abs(sourceTotalChecksum - targetTotalChecksum) / (double) sourceTotalChecksum;

            } catch (SQLException e) {
                throw new MigrationException("Data content comparison failed", e);
            }
        }

        private Object[] executeChecksumQuery(Connection conn, String sql, String merchantId) throws SQLException {
            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);
                try (ResultSet rs = stmt.executeQuery()) {
                    if (rs.next()) {
                        return new Object[]{
                                rs.getLong(1),
                                rs.getLong(2),
                                rs.getLong(3)
                        };
                    }
                    return new Object[]{0L, 0L, 0L};
                }
            }
        }

        private void cleanupData() {
            try (Connection conn = dataSource.getConnection()) {
                // 分批删除源表数据
                for (String merchantId : merchantIds) {
                    deleteMerchantData(conn, merchantId);
                }

                // 更新元数据状态
                completeMigration(conn);
            } catch (SQLException e) {
                // 清理错误可以记录但不需要失败
                log.error("Cleanup failed", e);
            }
        }
        private int calculateTotalBatches() {
            return merchantIds.stream()
                    .mapToInt(this::estimateBatchesForMerchant)
                    .sum();
        }

        private String generateInsertSQL(Connection conn, String tableName, Set<String> columns)
                throws SQLException
        {
            // 验证目标表存在
            if (!tableExists(conn, tableName)) {
                throw new MigrationException("Target table does not exist: " + tableName);
            }

            // 构造列名和占位符
            String colList = String.join(", ", columns);
            String placeholders = columns.stream().map(c -> "?").collect(Collectors.joining(", "));

            // 特殊处理：在冲突时忽略（防重插）
            return String.format("INSERT IGNORE INTO %s (%s) VALUES (%s)",
                    tableName, colList, placeholders);
        }
        private void handleMigrationError(Connection conn, Exception ex, int retryCount) {
            if (retryCount >= config.getMaxRetries()) {
                throw new MigrationException("Migration failed after " + retryCount + " retries", ex);
            }

            // 延迟退避重试
            long backoffMs = (long) (500 * Math.pow(2, retryCount));
            try {
                Thread.sleep(Math.min(backoffMs, 10000)); // 最大10秒
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }

            // 重置事务
            try {
                conn.rollback();
            } catch (SQLException e) {
                // 记录但忽略
            }
        }
        private void setParameter(PreparedStatement stmt, int index, Object value)
                throws SQLException
        {
            if (value == null) {
                stmt.setNull(index, Types.NULL);
            } else if (value instanceof Timestamp) {
                stmt.setTimestamp(index, (Timestamp) value);
            } else if (value instanceof java.util.Date) {
                stmt.setTimestamp(index, new Timestamp(((java.util.Date) value).getTime()));
            } else if (value instanceof Long) {
                stmt.setLong(index, (Long) value);
            } else if (value instanceof Integer) {
                stmt.setInt(index, (Integer) value);
            } else if (value instanceof String) {
                stmt.setString(index, (String) value);
            } else {
                // Fallback to object type
                stmt.setObject(index, value);
            }
        }

        private int estimateBatchesForMerchant(String merchantId) {
            long totalRecords = getRecordCount(merchantId);
            return (int) Math.ceil((double) totalRecords / config.getBatchSize());
        }
        private void rollbackMigration() {
            // 1. 设置回滚状态
            setState(MigrationState.ROLLBACK);

            try {
                // 2. 禁用双写模式
                disableDualWriting();

                // 3. 清理已迁移的数据
                cleanMigratedData();

                // 4. 删除新建的目标表
                dropTargetTable();

                // 5. 清理路由元数据
                cleanMigrationMetadata();

                // 6. 更新迁移状态为失败
                updateMigrationStatus("FAILED", "Rollback completed");

            } catch (MigrationException ex) {
                // 记录错误但继续执行清理
                log.error("Error during rollback execution", ex);
                updateMigrationStatus("FAILED", "Rollback with errors: " + ex.getMessage());
            } finally {
                // 7. 无论成功还是失败，最终状态都设置为回滚
                setState(MigrationState.ROLLBACK);

                // 8. 发送通知并释放锁
                sendRollbackNotification();
                releaseMigrationLock();
            }
        }
        private int calculateProgress(int processedBatches, int totalBatches) {
            if (totalBatches <= 0) return 100;
            double progress = (double) processedBatches / totalBatches * 100;
            return Math.min(100, (int) progress);
        }
        private long getMaxId(List<Map<String, Object>> batch) {
            // 假设批次中的最后一条记录具有最大的ID
            if (batch.isEmpty()) return 0;

            Map<String, Object> lastRow = batch.get(batch.size() - 1);
            Object idValue = lastRow.get("id");

            if (idValue instanceof Number) {
                return ((Number) idValue).longValue();
            } else {
                throw new MigrationException("ID column is not a number");
            }
        }

        private void updateMerchantStatus(Collection<String> merchantIds, String status) {
            String updateSql = "UPDATE merchant_status SET migration_status = ? WHERE merchant_id IN (?)";

            try (Connection conn = dataSource.getConnection()) {
                // 分批次更新，防止IN列表过长
                List<String> idList = new ArrayList<>(merchantIds);
                int chunkSize = 1000; // 每次更新1000个
                int chunks = (int) Math.ceil((double) idList.size() / chunkSize);

                for (int i = 0; i < chunks; i++) {
                    List<String> chunk = idList.subList(i * chunkSize, Math.min(i * chunkSize + chunkSize, idList.size()));

                    String inPart = chunk.stream()
                            .map(id -> "'" + id.replace("'", "''") + "'") // 简单防注入
                            .collect(Collectors.joining(","));

                    String sql = updateSql.replace("(?)", "(" + inPart + ")");

                    try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                        stmt.setString(1, status);
                        int updated = stmt.executeUpdate();
                        log.debug("Updated {} merchants to status: {}", updated, status);
                    }
                }
            } catch (SQLException e) {
                throw new MigrationException("Failed to update merchant status", e);
            }
        }

        private void disableDualWriting() {
            log.info("Disabling dual writing for merchants: {}", merchantIds);

            for (String merchantId : merchantIds) {
                try {
                    // 移除商户的双写状态
                    updateMerchantStatus(merchantId, "ACTIVE");

                    // 验证双写是否已禁用
                    if (isDualWritingEnabled(merchantId)) {
                        log.warn("Dual writing still enabled for merchant: {}", merchantId);
                    }
                } catch (Exception e) {
                    // 标记为需要手动检查
                    updateMerchantStatus(merchantId, "MANUAL_CHECK_REQUIRED");
                    throw new MigrationException("Failed to disable dual writing for merchant: " + merchantId, e);
                }
            }
        }
        private void sendRollbackNotification() {
            String subject = "Migration Rollback: " + taskId;
            String message = String.format(
                    "Migration from %s to %s for %d merchants has been rolled back.",
                    sourceTable, targetTable, merchantIds.size()
            );

            System.out.println(subject);
            System.out.println(message);
        }

        private void releaseMigrationLock() {
            try {
                lockManager.releaseLightLock(taskId);
            } catch (Throwable e) {
                // 锁释放失败，计划自动清理
               // scheduler.scheduleCleanup(taskId, Duration.ofHours(1));
                log.warn("Lock release failed. Scheduling cleanup for task: {}", taskId);
            }
        }

        private void checkPausedOrCancelled() {
            while (paused) {
                try {
                    // 每500ms检查一次恢复
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new MigrationException("Migration interrupted", e);
                }
            }

            if (cancelled) {
                throw new MigrationCancellationException("Migration was cancelled");
            }
        }
        private long getInitialMaxId(String merchantId) {
            String query = "SELECT max_id FROM migration_progress " +
                    "WHERE merchant_id = ? AND migration_id = ?";

            try (Connection conn = dataSource.getConnection();
                 PreparedStatement stmt = conn.prepareStatement(query)) {
                stmt.setString(1, merchantId);
                stmt.setString(2, taskId);

                try (ResultSet rs = stmt.executeQuery()) {
                    if (rs.next()) {
                        return rs.getLong("max_id");
                    }
                }
            } catch (SQLException e) {
                log.warn("Failed to fetch initial max ID for merchant: {}, starting from 0", merchantId);
            }
            return 0;
        }
        private void disableDualWriting() {
            log.info("Disabling dual writing for merchants: {}", merchantIds);

            for (String merchantId : merchantIds) {
                try {
                    // 移除商户的双写状态
                    updateMerchantStatus(merchantId, "ACTIVE");

                    // 验证双写是否已禁用
                    if (isDualWritingEnabled(merchantId)) {
                        log.warn("Dual writing still enabled for merchant: {}", merchantId);
                    }
                } catch (Exception e) {
                    // 标记为需要手动检查
                    updateMerchantStatus(merchantId, "MANUAL_CHECK_REQUIRED");
                    throw new MigrationException("Failed to disable dual writing for merchant: " + merchantId, e);
                }
            }
        }
        private void dropTargetTable() {
            if (tableExists(targetTable)) {
                log.info("Dropping target table: {}", targetTable);

                try (Connection conn = dataSource.getConnection()) {
                    // 安全删除表（仅当存在时）
                    String sql = "DROP TABLE IF EXISTS " + targetTable;
                    try (Statement stmt = conn.createStatement()) {
                        stmt.execute(sql);

                        // 检查表是否被真正删除
                        if (tableExists(targetTable)) {
                            log.error("Target table still exists: {}", targetTable);
                            throw new MigrationException("Failed to drop target table");
                        }
                    }
                } catch (SQLException e) {
                    throw new MigrationException("Table drop failed", e);
                }
            } else {
                log.info("Target table doesn't exist: {}", targetTable);
            }
        }

        private void cleanMigrationMetadata() {
            try (Connection conn = dataSource.getConnection()) {
                // 迁移记录标记为删除
                String markSql = "UPDATE migration_metadata " +
                        "SET deleted = true, clean_time = ? " +
                        "WHERE task_id = ?";
                try (PreparedStatement stmt = conn.prepareStatement(markSql)) {
                    stmt.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
                    stmt.setString(2, taskId);
                    stmt.executeUpdate();
                }

                // 移除分区锁
                String lockSql = "DELETE FROM partition_locks " +
                        "WHERE resource_type = ? AND resource_id = ?";
                try (PreparedStatement stmt = conn.prepareStatement(lockSql)) {
                    stmt.setString(1, "MERCHANT_MIGRATION");
                    stmt.setString(2, sourceTable + "_" + targetTable);
                    int deleted = stmt.executeUpdate();
                    log.info("Removed {} partition locks", deleted);
                }

                // 额外清理：删除临时文件/缓存
                deleteMigrationTempFiles();

            } catch (SQLException e) {
                log.error("Metadata cleanup failed", e);
                // 不中断整体回滚
            }
        }

        private void updateRoutingTable(String merchantId, String targetTable) {
            try (Connection conn = dataSource.getConnection()) {
                // 原子操作：更新路由并添加历史记录
                String transactionSQL =
                        "START TRANSACTION; " +
                                "INSERT INTO routing_history (merchant_id, old_table, new_table) " +
                                "SELECT merchant_id, table_name, ? " +
                                "FROM routing_table " +
                                "WHERE merchant_id = ?; " +
                                "UPDATE routing_table SET table_name = ? " +
                                "WHERE merchant_id = ?; " +
                                "COMMIT;";

                try (PreparedStatement stmt = conn.prepareStatement(transactionSQL)) {
                    stmt.setString(1, targetTable);
                    stmt.setString(2, merchantId);
                    stmt.setString(3, targetTable);
                    stmt.setString(4, merchantId);

                    stmt.executeUpdate();
                }

                // 验证更新
                if (!isRoutingCorrect(merchantId, targetTable)) {
                    throw new MigrationException("Routing update failed for merchant: " + merchantId);
                }

                log.info("Routing updated for merchant {} to table {}", merchantId, targetTable);
            } catch (SQLException e) {
                throw new MigrationException("Routing table update failed", e);
            }
        }

        // 路由正确性验证
        private boolean isRoutingCorrect(String merchantId, String expectedTable) {
            String sql = "SELECT table_name FROM routing_table WHERE merchant_id = ?";

            try (Connection conn = dataSource.getConnection();
                 PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);

                try (ResultSet rs = stmt.executeQuery()) {
                    return rs.next() && rs.getString(1).equals(expectedTable);
                }
            } catch (SQLException e) {
                log.error("Routing verification failed", e);
                return false;
            }
        }


        private void deleteMerchantData(Connection conn, String merchantId) throws SQLException {
            // 使用批处理删除 + 分块避免大事务
            long deleteThreshold = config.getBatchSize() * 100; // 10万条
            long totalRecords = getRecordCount(conn, sourceTable, merchantId);

            // 小数据量：单次删除
            if (totalRecords < deleteThreshold) {
                deleteSingleBatch(conn, merchantId);
            }
            // 大数据量：分块删除
            else {
                deleteInChunks(conn, merchantId);
            }
        }

        private void deleteSingleBatch(Connection conn, String merchantId) throws SQLException {
            String sql = "DELETE FROM " + sourceTable + " WHERE merchant_id = ?";

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);
                int deleted = stmt.executeUpdate();
                log.info("Deleted {} records for merchant: {}", deleted, merchantId);
            }
        }

        private void deleteInChunks(Connection conn, String merchantId) throws SQLException {
            long lastId = 0;
            int chunkSize = config.getBatchSize() * 10; // 5000条/块
            int totalDeleted = 0;

            String sql = "DELETE FROM " + sourceTable +
                    " WHERE merchant_id = ? AND id > ? AND id <= ? " +
                    "ORDER BY id LIMIT ?";

            while (true) {
                try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                    stmt.setString(1, merchantId);
                    stmt.setLong(2, lastId);
                    stmt.setLong(3, lastId + chunkSize);
                    stmt.setInt(4, chunkSize);

                    int deleted = stmt.executeUpdate();
                    if (deleted == 0) break;

                    totalDeleted += deleted;
                    lastId += chunkSize;

                    // 提交小事务避免锁表
                    conn.commit();

                    // 性能调节：避免过度消耗IO
                    Thread.sleep(50);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    throw new MigrationException("Deletion interrupted");
                }
            }

            log.info("Deleted {} records in chunks for merchant: {}", totalDeleted, merchantId);
        }

        private void completeMigration(Connection conn) throws SQLException {
            // 标记迁移成功
            String updateSQL =
                    "UPDATE migration_metadata SET " +
                            "status = 'COMPLETED', " +
                            "end_time = NOW(), " +
                            "record_count = (SELECT COUNT(*) FROM " + targetTable +
                            " WHERE merchant_id IN (" + getMerchantIdsPlaceholder() + ")) " +
                            "WHERE task_id = ?";

            try (PreparedStatement stmt = conn.prepareStatement(updateSQL)) {
                // 设置商户ID参数
                int index = 1;
                for (String merchantId : merchantIds) {
                    stmt.setString(index++, merchantId);
                }
                stmt.setString(index, taskId);

                int updated = stmt.executeUpdate();
                if (updated != 1) {
                    throw new MigrationException("Migration complete update failed");
                }
            }

            // 清理临时资源
            deleteMigrationTempFiles();

            // 释放数据库锁
            releaseMigrationLock(conn);

            // 发送成功通知
            sendMigrationSuccessNotification();

            log.info("Migration {} completed successfully", taskId);
        }

        // 生成商户ID占位符
        private String getMerchantIdsPlaceholder() {
            return merchantIds.stream()
                    .map(id -> "?")
                    .collect(Collectors.joining(","));
        }

        // 迁移成功通知
        private void sendMigrationSuccessNotification() {
            String subject = "Migration Completed: " + taskId;
            String body = String.format(
                    "Migration from %s to %s for %d merchants succeeded.%n" +
                            "Total records migrated: %,d%n" +
                            "Duration: %s",
                    sourceTable, targetTable, merchantIds.size(),
                    getMigratedRecordCount(),
                    Duration.between(startTime, LocalDateTime.now())
            );

            notificationService.sendAlert(subject, body);
        }

        private long getRecordCount(Connection conn, String tableName, String merchantId) throws SQLException {
            // 使用高精度COUNT(*)方法
            String countSQL = "SELECT COUNT(*) FROM " + tableName + " WHERE merchant_id = ?";

            try (PreparedStatement stmt = conn.prepareStatement(countSQL)) {
                stmt.setString(1, merchantId);

                try (ResultSet rs = stmt.executeQuery()) {
                    return rs.next() ? rs.getLong(1) : 0;
                }
            }
        }

        // 批量查询版本（提高效率）
        private Map<String, Long> getRecordCounts(Connection conn, String tableName) throws SQLException {
            String sql = "SELECT merchant_id, COUNT(*) AS cnt " +
                    "FROM " + tableName + " " +
                    "WHERE merchant_id IN (" + getMerchantIdsPlaceholder() + ") " +
                    "GROUP BY merchant_id";

            Map<String, Long> counts = new HashMap<>();

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                int index = 1;
                for (String merchantId : merchantIds) {
                    stmt.setString(index++, merchantId);
                }

                try (ResultSet rs = stmt.executeQuery()) {
                    while (rs.next()) {
                        counts.put(rs.getString("merchant_id"), rs.getLong("cnt"));
                    }
                }
            }

            // 为无数据的商户填0
            merchantIds.forEach(id -> counts.putIfAbsent(id, 0L));

            return counts;
        }

        private void cleanMigratedData() {
            try (Connection conn = dataSource.getConnection()) {
                // 为每个商户生成清理任务
                merchantIds.forEach(merchantId -> cleanMerchantData(conn, merchantId));

                // 最终验证
                Map<String, Long> remainingCounts = getRecordCounts(conn, targetTable);
                long totalRemaining = remainingCounts.values().stream().mapToLong(Long::longValue).sum();

                if (totalRemaining > 0) {
                    log.error("Found {} remaining records after cleanup", totalRemaining);
                    throw new MigrationException("Data cleanup incomplete");
                }
            } catch (SQLException e) {
                throw new MigrationException("Data cleanup failed", e);
            }
        }

        private void cleanMerchantData(Connection conn, String merchantId) {
            // 分阶段删除
            try {
                // 阶段1：删除95%的数据
                int batchSize = config.getBatchSize();
                long totalRecords = getRecordCount(conn, targetTable, merchantId);

                // 批量删除循环
                while (getRecordCount(conn, targetTable, merchantId) > 0) {
                    deleteBatch(conn, merchantId, batchSize);

                    // 性能调控：每删100批休息1秒
                    if (++batchCount % 100 == 0) {
                        Thread.sleep(1000);
                    }
                }

                log.info("Data cleaned for merchant: {}", merchantId);
            } catch (Exception e) {
                log.error("Cleanup failed for merchant: {}", merchantId, e);
                throw new MigrationException("Partial cleanup failure", e);
            }
        }

        private void deleteBatch(Connection conn, String merchantId, int batchSize) throws SQLException {
            // 使用子查询删除优化性能
            String sql =
                    "DELETE FROM " + targetTable +
                            " WHERE merchant_id = ? AND id IN (" +
                            "    SELECT id FROM (" +
                            "        SELECT id FROM " + targetTable +
                            "        WHERE merchant_id = ?" +
                            "        ORDER BY id LIMIT ?" +
                            "    ) AS sub" +
                            ")";

            try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                stmt.setString(1, merchantId);
                stmt.setString(2, merchantId);
                stmt.setInt(3, batchSize);

                int deleted = stmt.executeUpdate();
                if (deleted > 0) {
                    conn.commit(); // 提交小事务
                }
            }
        }

        private void deleteMigrationTempFiles() {
            Path tempDir = Paths.get(config.getTempDir(), taskId);

            try {
                // 递归删除临时目录
                Files.walkFileTree(tempDir, new SimpleFileVisitor<Path>() {
                    @Override
                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
                            throws IOException {
                        Files.delete(file);
                        return FileVisitResult.CONTINUE;
                    }

                    @Override
                    public FileVisitResult postVisitDirectory(Path dir, IOException exc)
                            throws IOException {
                        Files.delete(dir);
                        return FileVisitResult.CONTINUE;
                    }
                });
                log.info("Deleted temp files for migration: {}", taskId);
            } catch (IOException e) {
                log.error("Failed to delete temp files", e);
            }
        }

        private void deleteMigrationTempFiles() {
            Path tempDir = Paths.get(config.getTempDir(), taskId);

            try {
                // 递归删除临时目录
                Files.walkFileTree(tempDir, new SimpleFileVisitor<Path>() {
                    @Override
                    public FileVisitResult visitFile(Path file, BasicFileAttributes attrs)
                            throws IOException {
                        Files.delete(file);
                        return FileVisitResult.CONTINUE;
                    }

                    @Override
                    public FileVisitResult postVisitDirectory(Path dir, IOException exc)
                            throws IOException {
                        Files.delete(dir);
                        return FileVisitResult.CONTINUE;
                    }
                });
                log.info("Deleted temp files for migration: {}", taskId);
            } catch (IOException e) {
                log.error("Failed to delete temp files", e);
            }
        }



        private void updateMigrationStatus(String status, String reason) {
            try (Connection conn = dataSource.getConnection()) {
                String sql = "UPDATE migration_metadata " +
                        "SET status = ?, end_time = ?, reason = ? " +
                        "WHERE task_id = ?";

                try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                    stmt.setString(1, status);
                    stmt.setTimestamp(2, new Timestamp(System.currentTimeMillis()));
                    stmt.setString(3, reason);
                    stmt.setString(4, taskId);

                    int updated = stmt.executeUpdate();
                    if (updated != 1) {
                        log.warn("Failed to update migration status for task: {}", taskId);
                    }
                }
            } catch (SQLException e) {
                log.error("Status update failed", e);
            }
        }

        private boolean isDualWritingEnabled(String merchantId) {
            try (Connection conn = dataSource.getConnection()) {
                String sql = "SELECT dual_write_enabled FROM merchant_status WHERE merchant_id = ?";
                try (PreparedStatement stmt = conn.prepareStatement(sql)) {
                    stmt.setString(1, merchantId);
                    try (ResultSet rs = stmt.executeQuery()) {
                        return rs.next() && rs.getBoolean("dual_write_enabled");
                    }
                }
            } catch (SQLException e) {
                log.error("Failed to check dual write status", e);
                return true; // 保守策略：默认为开启，需要进一步处理
            }
        }


        private void createTempTableForCleanup(Connection conn) throws SQLException {
            String tempTableName = "cleanup_" + taskId.replace("-", "_");

            // 创建临时表（自动清理）
            String createSQL = "CREATE TEMPORARY TABLE " + tempTableName + " (" +
                    "id BIGINT PRIMARY KEY," +
                    "merchant_id VARCHAR(20) NOT NULL" +
                    ") ENGINE=MEMORY";

            try (Statement stmt = conn.createStatement()) {
                stmt.execute(createSQL);
            }

            // 填充待清理的ID
            String insertSQL = "INSERT INTO " + tempTableName + " " +
                    "SELECT id, merchant_id FROM " + targetTable + " " +
                    "WHERE merchant_id IN (" +
                    String.join(",", Collections.nCopies(merchantIds.size(), "?")) +
                    ")";

            try (PreparedStatement stmt = conn.prepareStatement(insertSQL)) {
                int index = 1;
                for (String merchantId : merchantIds) {
                    stmt.setString(index++, merchantId);
                }
                stmt.execute();
            }
        }


        private void deleteBatch(Connection conn, List<Long> ids) throws SQLException {
            // 构建参数化删除语句
            String placeholders = ids.stream().map(i -> "?").collect(Collectors.joining(","));
            String deleteSQL = "DELETE FROM " + targetTable + " WHERE id IN (" + placeholders + ")";

            try (PreparedStatement stmt = conn.prepareStatement(deleteSQL)) {
                int index = 1;
                for (Long id : ids) {
                    stmt.setLong(index++, id);
                }
                int deleted = stmt.executeUpdate();

                if (deleted != ids.size()) {
                    log.warn("Inconsistent delete: expected {}, actually {}", ids.size(), deleted);
                }
            }
        }

        private boolean tableExists(String tableName) {
            try (Connection conn = dataSource.getConnection()) {
                DatabaseMetaData meta = conn.getMetaData();
                try (ResultSet rs = meta.getTables(null, null, tableName, new String[]{"TABLE"})) {
                    return rs.next();
                }
            } catch (SQLException e) {
                log.error("Table existence check failed", e);
                return false;
            }
        }


        // 辅助方法实现略（篇幅考虑）
    }

    // 轻量级锁管理器
    public static class LightLockManager {
        private final Map<String, LockEntry> lockMap = new ConcurrentHashMap<>();
        private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor();

        public LightLockManager() {
            cleaner.scheduleAtFixedRate(this::cleanExpiredLocks, 1, 1, TimeUnit.MINUTES);
        }

        public boolean acquireLightLock(String resourceId, Duration duration) {
            LockEntry newEntry = new LockEntry(duration);
            LockEntry existing = lockMap.putIfAbsent(resourceId, newEntry);

            if (existing != null && existing.isValid()) {
                return false; // 锁被占用
            }

            lockMap.put(resourceId, newEntry);
            return true;
        }

        public void releaseLightLock(String resourceId) {
            lockMap.remove(resourceId);
        }

        private void cleanExpiredLocks() {
            lockMap.entrySet().removeIf(entry -> !entry.getValue().isValid());
        }

        private static class LockEntry {
            final LocalDateTime expireTime;

            LockEntry(Duration duration) {
                this.expireTime = LocalDateTime.now().plus(duration);
            }

            boolean isValid() {
                return LocalDateTime.now().isBefore(expireTime);
            }
        }
    }

    // 迁移服务入口
    public static class MigrationService {
        private final LightLockManager lockManager = new LightLockManager();
        private final DataSource dataSource;
        private final MigrationConfig defaultConfig = new MigrationConfig();
        private final ExecutorService executor = Executors.newCachedThreadPool();
        private final Map<String, DataMigrator> activeMigrations = new ConcurrentHashMap<>();

        public MigrationService(DataSource dataSource) {
            this.dataSource = dataSource;
        }

        public String startMigration(Set<String> merchantIds, String sourceTable,
                                     String targetTable, MigrationConfig config) {

            MigrationConfig effectiveConfig = config != null ? config : defaultConfig;
            String migrationId = UUID.randomUUID().toString();

            DataMigrator migrator = new DataMigrator(
                    migrationId, merchantIds, sourceTable, targetTable, effectiveConfig
            );

            // 添加监控监听器
            migrator.addListener(new LoggingMigrationListener());

            activeMigrations.put(migrationId, migrator);
            executor.submit(migrator);

            return migrationId;
        }

        public void pauseMigration(String migrationId) {
            DataMigrator migrator = activeMigrations.get(migrationId);
            if (migrator != null) {
                migrator.pause();
            }
        }

        public void resumeMigration(String migrationId) {
            DataMigrator migrator = activeMigrations.get(migrationId);
            if (migrator != null) {
                migrator.resume();
            }
        }

        public void cancelMigration(String migrationId) {
            DataMigrator migrator = activeMigrations.get(migrationId);
            if (migrator != null) {
                migrator.cancel();
            }
        }

        public MigrationState getMigrationState(String migrationId) {
            DataMigrator migrator = activeMigrations.get(migrationId);
            return migrator != null ? migrator.getState() : null;
        }

        public int getMigrationProgress(String migrationId) {
            DataMigrator migrator = activeMigrations.get(migrationId);
            return migrator != null ? migrator.getProgress() : -1;
        }
    }

    private static final Logger log = LoggerFactory.getLogger("test");
    // 日志监听器实现
    private static class LoggingMigrationListener implements MigrationListener {
        @Override
        public void onStateChanged(MigrationState newState) {
            log.info("Migration state changed to: " + newState);
        }

        @Override
        public void onProgressUpdate(int progress) {
            log.info("Migration progress: {}%", progress);
        }

        @Override
        public void onError(Throwable exception) {
            log.error("Migration error", exception);
        }
    }

    // 异常类
    public static class MigrationException extends RuntimeException {
        public MigrationException(String message) {
            super(message);
        }

        public MigrationException(String message, Throwable cause) {
            super(message, cause);
        }
    }
}